/**
 * 
 */
package io.gitee.falllee.dao.core.table;

import java.util.HashMap;
import java.util.List;
import java.util.Map;

import org.springframework.core.env.Environment;
import org.springframework.util.StringUtils;

import io.gitee.falllee.dao.core.HeaderBuilder;
import io.gitee.falllee.dao.enums.EnableConfigEnum;
import io.gitee.falllee.dao.model.DaoConst;
import io.gitee.falllee.kafka.util.KafkaMessageUtils;

/**
 * @author batty
 * @date 2022-06-27
 *
 */
public class MessageBuilder extends HeaderBuilder {
    
    private String tableName;
    
    private Integer kafka;
    
    MessageBuilder(TableBuilder parent) {
        super(parent.getDaoConfig());
        this.tableName = parent.getTableName();
        this.kafka = parent.getKafka();
    }
    
    public MessageBuilder header(String key, String value) {
        if (headerParamsMap == null) {
            headerParamsMap = new HashMap<>(16);
        }
        headerParamsMap.put(key, value);
        return this;
    }

    public void sendMessage(Object keyValue, String op) {
        if (enableKafka()) {
            String key = getMessageKey(op);
            Map<String, String> headerParamsMap = getHeaderParamsMap();
            String topic = getConfigTopic();
            KafkaMessageUtils.sendMessage(headerParamsMap, topic, key, keyValue);
        }
    }

    public void sendMessage(List<Object> keyList, String op) {
        if (enableKafka()) {
            String key = getMessageKey(op);
            Map<String, String> headerParamsMap = getHeaderParamsMap();
            String topic = getConfigTopic();
            KafkaMessageUtils.sendMessage(headerParamsMap, topic, key, keyList);
        }
    }
    
    /**
     * 是否启用Kafka消息(优先局部配置；如果局部配置为0，则使用全局配置；如果都为0则返回false)
     * 
     * @return 是否启用
     */
    private boolean enableKafka() {
        if (EnableConfigEnum.ENABLE.getConfig().equals(kafka)) {
            return true;
        } else if (EnableConfigEnum.DISABLE.getConfig().equals(kafka)) {
            return false;
        }
        if (EnableConfigEnum.ENABLE.getConfig().equals(daoConfig.getEnableKafka())) {
            return true;
        } else if (EnableConfigEnum.DISABLE.getConfig().equals(daoConfig.getEnableKafka())) {
            return false;
        }
        return false;
    }

    /**
     * 获取Kafka消息的Key
     * 
     * @param clazz 实体类型
     * @return 消息Key
     */
    private String getMessageKey(String op) {
        StringBuilder sb = new StringBuilder();
        Environment env = daoConfig.getEnvironment();
        String msName = env.getProperty(DaoConst.SERVICE_NAME_CONFIG);
        if (StringUtils.hasLength(msName)) {
            sb.append(msName);
            sb.append(DaoConst.SEPARATOR);
        }
        sb.append(op);
        sb.append(DaoConst.SEPARATOR);
        sb.append(tableName);
        return sb.toString();
    }

}
